package net.bwie.realtime.jtp.dwd.logs.car.data;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;
public class VehicleDataProducer {
    private static final String TOPIC = "vehicle_realtime";
    private static final String BOOTSTRAP_SERVERS = "node101:9092";

    private static final Random random = new Random();
    private static final ObjectMapper mapper = new ObjectMapper();

    private static final String VIN_PREFIX = "LJW"; // 固定前缀
    private static final int VIN_LENGTH = 17;
    private static final long START = 1;   // 范围起始
    private static final long END = 10000;   // 范围结束（比如 100 辆车）

    // VIN 范围生成方法
    private static String generateVinInRange() {
        long number = START + random.nextInt((int) (END - START + 1));
        String numberStr = String.format("%0" + (VIN_LENGTH - VIN_PREFIX.length()) + "d", number);
        return VIN_PREFIX + numberStr;
    }

    // 模拟生成车辆实时数据
    private static VehicleData generateRandomVehicleData() {
        VehicleData data = new VehicleData();
        data.setVin(generateVinInRange());
        data.setTimestamp(System.currentTimeMillis());
        data.setType(random.nextInt(2)); // 0=启动, 1=熄火
        data.setLongitude(round(100 + random.nextDouble() * 20));
        data.setLatitude(round(20 + random.nextDouble() * 20));
        data.setSpeed(round(random.nextDouble() * 200));
        data.setMileage(round(5000 + random.nextDouble() * 100000));
        data.setBatteryLevel(round(random.nextDouble() * 100));
        data.setBatteryTemp(round(20 + random.nextDouble() * 40));
        data.setMotorTemp(round(30 + random.nextDouble() * 50));
        data.setChargingStatus(random.nextBoolean() ? "charging" : "discharging");
        data.setEnergyConsumption(round(random.nextDouble() * 30));
        return data;
    }

    // 保留两位小数
    private static double round(double value) {
        return Math.round(value * 100.0) / 100.0;
    }

    public static void main(String[] args) throws Exception {
        // Kafka 配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            while (true) {
                VehicleData vehicleData = generateRandomVehicleData();
                String json = mapper.writeValueAsString(vehicleData);

                ProducerRecord<String, String> record =
                        new ProducerRecord<>(TOPIC, vehicleData.getVin(), json);

                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        System.out.printf("消息已发送: partition=%d, offset=%d, value=%s%n",
                                metadata.partition(), metadata.offset(), json);
                    } else {
                        exception.printStackTrace();
                    }
                });

                Thread.sleep(1000);
            }
        } finally {
            producer.close();
        }
    }


    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    private static class VehicleData {
        private String vin;  // 车辆唯一标识
        private Long timestamp; // 数据时间戳
        private int type; // 运行状态 0=启动,1=熄火
        private Double longitude; // 经度
        private Double latitude;  // 纬度
        private Double speed;     // 车速
        private Double mileage;   // 总里程
        private Double batteryLevel; // 电池电量
        private Double batteryTemp;  // 电池温度
        private Double motorTemp;    // 电机温度
        private String chargingStatus; // 充电状态
        private Double energyConsumption; // 瞬时能耗
    }
}
